package com.zhss.dfs.namenode.server;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.zhss.dfs.namenode.rpc.model.*;
import com.zhss.dfs.namenode.rpc.service.*;

import com.zhss.dfs.namenode.server.context.DfsEnv;
import com.zhss.dfs.namenode.server.context.Environment;
import com.zhss.dfs.namenode.server.editlog.FSEditlog;
import io.grpc.stub.StreamObserver;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.List;

/**
 * NameNode的rpc服务的接口
 *
 * @author zhonghuashishan
 */
public class NameNodeServiceImpl implements NameNodeServiceGrpc.NameNodeService {

    public static final Integer STATUS_SUCCESS = 1;
    public static final Integer STATUS_FAILURE = 2;
    public static final Integer STATUS_SHUTDOWN = 3;
    public static final Integer BACKUP_NODE_FETCH_SIZE = 10;
    public Environment env = DfsEnv.getInstance();
    //当前缓冲的一小部分editlog
    private JSONArray currentBufferedEditLog = new JSONArray();
    /**
     * 负责管理元数据的核心组件
     */
    private FSNameSystem namesystem;
    /**
     * 负责管理集群中所有的datanode的组件
     */
    private DataNodeManager datanodeManager;
    //是否还在运行
    private volatile boolean isRunning = true;

    //当前内存里缓冲了哪个磁盘文件的数据
    private String bufferedFlushedTxid;
    // 当前内存缓存里最大的txid;
    private long currentBufferMaxTxid = 0l;

    public NameNodeServiceImpl(
            FSNameSystem namesystem,
            DataNodeManager datanodeManager) {
        this.namesystem = namesystem;
        this.datanodeManager = datanodeManager;
    }

    /**
     * 创建目录
     *
     * @param path 目录路径
     * @return 是否创建成功
     * @throws Exception
     */
    public Boolean mkdir(String path) throws Exception {
        return this.namesystem.mkdir(path);
    }

    /**
     * datanode进行注册
     *
     * @return
     * @throws Exception
     */

    @Override
    public void register(RegisterRequest request,
                         StreamObserver<RegisterResponse> responseObserver) {
        datanodeManager.register(request.getIp(), request.getHostname());

        RegisterResponse response = RegisterResponse.newBuilder()
                .setStatus(STATUS_SUCCESS)
                .build();

        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    /**
     * datanode进行心跳
     *
     * @return
     * @throws Exception
     */
    @Override
    public void heartbeat(HeartbeatRequest request,
                          StreamObserver<HeartbeatResponse> responseObserver) {
        datanodeManager.heartbeat(request.getIp(), request.getHostname());

        HeartbeatResponse response = HeartbeatResponse.newBuilder()
                .setStatus(STATUS_SUCCESS)
                .build();

        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    /**
     * 创建目录
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void mkdir(MkdirRequest request, StreamObserver<MkdirResponse> responseObserver) {
        try {
            if (!isRunning) {
                responseObserver.onNext(MkdirResponse.newBuilder().setStatus(STATUS_SHUTDOWN).build());
            } else {
                this.namesystem.mkdir(request.getPath());
                responseObserver.onNext(MkdirResponse.newBuilder().setStatus(STATUS_SUCCESS).build());
            }
            responseObserver.onCompleted();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 优雅关闭接口
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void shutdown(ShutdownRequest request, StreamObserver<ShutdownResponse> responseObserver) {
        this.isRunning = false;
        this.namesystem.flush();
        namesystem.saveCheckpointTxid();
        responseObserver.onCompleted();
    }

    @Override
    public void fetchEditsLog(FetchEditsLogRequest request, StreamObserver<FetchEditsLogResponse> responseObserver) {
        if (!isRunning) {
            FetchEditsLogResponse response = FetchEditsLogResponse.newBuilder()
                    .setEditsLog(new JSONArray().toJSONString())
                    .build();
            responseObserver.onNext(response);
            responseObserver.onCompleted();
            return;
        }
        FetchEditsLogResponse response = null;
        long syncedTxid = request.getSyncedTxid();
        FSEditlog editLog = this.namesystem.getEditlog();
        List<String> flushedTxids = editLog.getFlushedTxids();
        JSONArray fetchedEditLog = new JSONArray();
        //如果此时还没有刷出来任何磁盘文件的话，那么此时数据仅仅存在于内存缓冲里
        if (flushedTxids.size() == 0) {
            System.out.println("暂时没有任何磁盘文件，直接从内存缓冲中拉取editlog...");
            fetchFromBufferEditsLog(fetchedEditLog, syncedTxid);
        } else {
            //发现已经有落地磁盘的文件了，这个时候就要扫描所有磁盘文件的索引范围
            //第一种情况 要拉取的txid是在某个磁盘文件里
            //判断是否再缓存里
            if (bufferedFlushedTxid != null) {
                if (existInDiskFile(bufferedFlushedTxid, syncedTxid)) {
                    System.out.println("上一次已经存在过磁盘文件的数据，直接从磁盘文件缓存中拉取editlog....");
                    fetchFromCurrentBuffer(fetchedEditLog, syncedTxid);
                } else {
                    //如果要拉取的数据不在当前缓存的磁盘文件里了，那么需要从下一个磁盘文件去拉
                    String nextFlushedTxid = getNextFlushedTxid(flushedTxids);
                    if (nextFlushedTxid != null) {
                        System.out.println("上一次缓存的磁盘文件找不到要拉取的数据，从下一个磁盘文件中拉取editlog...");
                        fetchFromFlushedFile(nextFlushedTxid, fetchedEditLog, syncedTxid);
                    } else {
                        //没有读到下一个文件，此时需要从内存里读取
                        System.out.println("上一次缓存的磁盘文件找不到要拉取的数据，而且没有下一个磁盘文件，尝试从内存缓冲中拉取editlog...");
                        fetchFromBufferEditsLog(fetchedEditLog, syncedTxid);
                    }

                }
            }
            //第一次尝试从磁盘文件里去拉取
            else {
                boolean fetchFromFlushFileFlag = false;
                //遍历所有磁盘文件的索引范围
                for (String flushedTxid : flushedTxids) {
                    //如果此时要拉取的数据在某个磁盘里
                    if (existInDiskFile(flushedTxid, syncedTxid)) {
                        System.out.println("尝试从磁盘文件中拉去editlog,flushedTxid=" + flushedTxid);
                        fetchFromFlushedFile(flushedTxid, fetchedEditLog, syncedTxid);
                        fetchFromFlushFileFlag = true;
                        break;
                    }
                }
                if (!fetchFromFlushFileFlag) {
                    //第二种情况 要拉取的txid已经比磁盘文件里的全部都新了，还在内存缓冲里
                    System.out.println("所有磁盘都没找到要拉取的editlog直接从内存缓存中拉取editlog...");
                    fetchFromBufferEditsLog(fetchedEditLog, syncedTxid);
                }
            }
        }
        response = FetchEditsLogResponse.newBuilder().setEditsLog(fetchedEditLog.toJSONString()).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    @Override
    public void updateCheckpointTxid(UpdateCheckpointTxidRequest request, StreamObserver<UpdateCheckpointTxidResponse> responseObserver) {
        long txid = request.getTxid();
        namesystem.setCheckpointTxid(txid);
        UpdateCheckpointTxidResponse response = UpdateCheckpointTxidResponse.newBuilder()
                .setStatus(1)
                .build();

        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    private String getNextFlushedTxid(List<String> flushedTxids) {
        String nextFlushedTxid = null;
        for (int i = 0; i < flushedTxids.size(); i++) {
            if (flushedTxids.get(i).equals(bufferedFlushedTxid)) {
                if (i + 1 < flushedTxids.size()) {
                    nextFlushedTxid = flushedTxids.get(i + 1);

                }
            }
        }
        return nextFlushedTxid;
    }

    /**
     * 从内存缓冲的editlog中拉取数据
     *
     * @param fetchedEditLog
     */
    private void fetchFromBufferEditsLog(JSONArray fetchedEditLog, Long syncedTxid) {
        //如果要拉取的txid还在上一次的内存缓存中,此时继续从内存缓存中拉取即可
        long fetchTxid = syncedTxid + 1;
        if (fetchTxid <= currentBufferMaxTxid) {
            System.out.println("尝试从内存缓冲拉取的时候，发现上一次内存缓存里有数据可供拉取,直接从缓存里拉取editlog...");
            fetchFromCurrentBuffer(fetchedEditLog, syncedTxid);
            return;
        }

        currentBufferedEditLog.clear();
        //从内存里重新拉一次
        String[] bufferEditsLog = this.namesystem.getEditlog().getBufferEditsLog();
        if (bufferEditsLog != null) {
            for (String editlog : bufferEditsLog) {
                currentBufferedEditLog.add(JSONObject.parseObject(editlog));
                currentBufferMaxTxid = JSONObject.parseObject(editlog).getLong("txid");
                //在这里记录一下,当前内存缓冲中最大的txid,这样下次再拉取时可以判断，内存里的数据是否还可以再读取，不用每次去缓冲中加载
            }
            bufferedFlushedTxid = null;
            fetchFromCurrentBuffer(fetchedEditLog, syncedTxid);
            return;
        }

    }

    //从内存缓冲的editlog中拉取数据
    private void fetchFromCurrentBuffer(JSONArray fetchedEditLog, Long syncedTxid) {
        int fetchCount = 0;
        for (int i = 0; i < currentBufferedEditLog.size(); i++) {
            if (currentBufferedEditLog.getJSONObject(i).getLong("txid") > syncedTxid) {
                syncedTxid = currentBufferedEditLog.getJSONObject(i).getLong("txid");
                fetchedEditLog.add(currentBufferedEditLog.getJSONObject(i));
                fetchCount++;

                if (fetchCount >= BACKUP_NODE_FETCH_SIZE) {
                    break;
                }
            }
        }
    }

    private Boolean existInDiskFile(String flushedTxid, Long syncedTxid) {
        String[] flushedTxidSplit = flushedTxid.split("-");
        long startTxid = Long.valueOf(flushedTxidSplit[0]);
        long endTxid = Long.valueOf(flushedTxidSplit[1]);
        long fetchBeginTxid = syncedTxid + 1;
        if (fetchBeginTxid >= startTxid && fetchBeginTxid <= endTxid) {
            return true;
        }
        return false;

    }

    //从已经刷入磁盘的文件里读取editlog,同时缓存文件数据到内存
    public void fetchFromFlushedFile(String flushedTxid, JSONArray fetchedEditLog, Long syncedTxid) {

        currentBufferedEditLog.clear();

        String currentEditLogFile = env.getEditlogDirPath() + "/editlog/" + flushedTxid + ".log";
        try {
            List<String> editlogs = Files.readAllLines(Paths.get(currentEditLogFile), StandardCharsets.UTF_8);
            for (String editlog : editlogs) {
                currentBufferedEditLog.add(JSONObject.parseObject(editlog));
                currentBufferMaxTxid = JSONObject.parseObject(editlog).getLong("txid");
            }

            bufferedFlushedTxid = flushedTxid; //缓存了某个磁盘文件的标志位

            fetchFromCurrentBuffer(fetchedEditLog, syncedTxid);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


}
